#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include"muduo/net/TcpConnection.h"
#include"muduo/proto/codec.h"
#include<google/protobuf/map.h>

#include"../mqcommon/mq_msg.pb.h"
#include"../mqcommon/helper.hpp"
#include"../mqcommon/logger.hpp"
#include"../mqcommon/mq_proto.pb.h"
#include"../mqcommon/mq_threadpool.hpp"
#include"mq_virtualhost.hpp"
#include"mq_consumer.hpp"
#include"mq_route.hpp"

#include<functional>
#include<mutex>
#include<memory>

namespace mq{
    
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumerRequestPtr = std::shared_ptr<basicConsumerRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    

    
    class Channel{
        public:
            using ptr = std::shared_ptr<Channel>;
            Channel(const std::string& cid,
                    const ConsumerManager::ptr& consumerManagerP,
                    const muduo::net::TcpConnectionPtr& connectionP,
                    const ProtobufCodecPtr& codecP,
                    const Virtualhost::ptr& vHost,
                    const ThreadPool::ptr& pool):
                    _cid(cid),_consumerManagerP(consumerManagerP),
                    _connectionP(connectionP),_codecP(codecP),
                    _vHost(vHost),_pool(pool)
                    {}

            ~Channel(){
                if(_consumer.get() != nullptr){
                    _consumerManagerP->removeConsumer(_consumer->tag,_consumer->qname);
                }
            }
            
            void declareExchange(const declareExchangeRequestPtr& req){
                int ret = _vHost->declareExchange(req->exchange_name(),req->type()
                    ,req->durable(),req->auto_delete(),req->args());
                if(ret == false){
                    LOG_ERRORM("declareExchange false...%s",req->exchange_name().c_str());
                    return Channel::basicResponse(ret,req->rid(),req->cid());
                }
                return Channel::basicResponse(ret,req->rid(),req->cid());

            }

            void deleteExchange(const deleteExchangeRequestPtr& req){
                _vHost->deleteExchange(req->exchange_name());
                return Channel::basicResponse(true,req->rid(),req->cid());
            }
                
            void declareQueue(const declareQueueRequestPtr& req){
                int ret = _vHost->declareQueue(req->queue_name(),req->durable(),req->exclusive(),req->auto_delete(),req->args());
                if(ret == false){
                    LOG_ERRORM("declareQueue false...%s",req->queue_name().c_str());
                    return Channel::basicResponse(ret,req->rid(),req->cid());
                }
                ret = _consumerManagerP->initQueueConsumer(req->queue_name());
                if(ret == false){
                    LOG_ERRORM("initQueueConsumer false... Queue delete...");
                    _vHost->deleteQueue(req->queue_name());
                    return Channel::basicResponse(ret,req->rid(),req->cid());
                }
                return Channel::basicResponse(ret,req->rid(),req->cid());
            }
            //删除队列 和 解除绑定！
            void deleteQueue(const deleteQueueRequestPtr& req){
                _vHost->deleteQueue(req->queue_name());
                _consumerManagerP->destoryQueueConsumer(req->queue_name());
                return Channel::basicResponse(true,req->rid(),req->cid());
            }

            void bind(const queueBindRequestPtr& req){
                int ret = _vHost->bind(req->exchange_name(),req->queue_name(),req->binding_key());
                return Channel::basicResponse(ret,req->rid(),req->cid());
            }
            void unBind(const queueUnBindRequestPtr& req){
                _vHost->unBind(req->exchange_name(),req->queue_name());
                return Channel::basicResponse(true,req->rid(),req->cid());
                
            }

            void basicPublish(const basicPublishRequestPtr& req){
                //查看有无ename 遍历匹配
                mq::Exchange::ptr ep = _vHost->getExchange(req->exchange_name());
                if(ep.get() == nullptr){
                    LOG_ERRORM("没有相关交换机...%s",req->exchange_name().c_str());
                    return Channel::basicResponse(false,req->rid(),req->cid());
                }
                mq::MsgQueueBindMap mqbp =  _vHost->exchangeBings(req->exchange_name());
                BasicProperty* bp = nullptr;
                std::string routingkey;
                if(req->has_property()){
                    bp = req->mutable_property();
                    routingkey = req->property().routingkey();
                    int ret = Router::IsLegalRoutingkey(routingkey);
                    if(ret == false){
                        LOG_ERRORM("Is NO Legal Routingkey...");
                        return Channel::basicResponse(ret,req->rid(),req->cid());
                    }
                }
                for(auto& mqb : mqbp){
                    if(Router::route(ep->type,routingkey,mqb.second->binding_key)){
                        _vHost->basicPublish(mqb.first,bp,req->body());
                        auto func_push = std::bind(&Channel::consume,this,mqb.first);
                        _pool->push(func_push);
                    }
                }
                return Channel::basicResponse(true,req->rid(),req->cid());

            }
            void basicAck(const basicAckRequestPtr& req){
                _vHost->basicAck(req->queue_name(),req->message_id());
                return Channel::basicResponse(true,req->rid(),req->cid());
            }
            //订阅队列消息
            void basicConsume(const basicConsumerRequestPtr& req){
                // _vHost->consume()
                bool ret = _vHost->existsQueue(req->queue_name());
                if(ret == false){
                    LOG_ERRORM("no queue can't subscribe");
                    return Channel::basicResponse(ret,req->rid(),req->cid()); 
                }
                auto fun = std::bind(&Channel::ConsumerCallBack,this,std::placeholders::_1,
                std::placeholders::_2,std::placeholders::_3);
                //绑定 ConsumerCallBack
                _consumer =  _consumerManagerP->createConsumer(req->consumer_tag(),req->queue_name(),req->auto_ack(),fun);
                LOG_DEBUGM("basicConsume done...");
                return Channel::basicResponse(true,req->rid(),req->cid()); 
            }
            void basicCancel(const basicCancelRequestPtr& req){
                _consumerManagerP->removeConsumer(req->cid(),req->queue_name());
                return Channel::basicResponse(true,req->rid(),req->cid()); 
            }
        private:
            void ConsumerCallBack(const std::string& ctag,const BasicProperty* bp,const std::string& body){
                basicConsumerResponse rsp;
                rsp.set_cid(_cid);
                rsp.set_body(body);
                rsp.set_consumer_tag(ctag);
                if(bp != nullptr){
                    rsp.mutable_property()->set_id(bp->id());
                    rsp.mutable_property()->set_deliverymod(bp->deliverymod());
                    rsp.mutable_property()->set_routingkey(bp->routingkey());
                }
                _codecP->send(_connectionP,rsp);
            }

            void consume(const std::string& queue_name){
                mq::MqMessagePtr msg =  _vHost->consume(queue_name);
                if(msg.get() == nullptr){
                    LOG_ERRORM("queue %s 没有消息了",queue_name.c_str());
                    return ;
                }
                mq::Consumer::ptr consumer = _consumerManagerP->choose(queue_name);
                if(consumer.get() == nullptr){
                    // LOG_ERRORM("queue %s 没有消费者",queue_name.c_str());
                    return ;
                }
                //运行ConsumerCallBack
                consumer->cb(consumer->tag,msg->mutable_payload()->mutable_basepro(),msg->payload().body());
                if(consumer->auto_ack) _vHost->basicAck(queue_name,msg->payload().basepro().id());
            }

            void basicResponse(bool ok,std::string rid,std::string cid){
                basicCommonResponse rep;
                rep.set_ok(ok);
                rep.set_rid(rid);
                rep.set_cid(cid);
                _codecP->send(_connectionP,rep);
            }

        private:
            std::string _cid;
            Consumer::ptr _consumer;
            ConsumerManager::ptr _consumerManagerP;
            muduo::net::TcpConnectionPtr _connectionP;
            ProtobufCodecPtr _codecP;
            Virtualhost::ptr _vHost;
            ThreadPool::ptr _pool;
    };
    class ChannelManager{
        public:
            using ptr = std::shared_ptr<ChannelManager>;

            ChannelManager(){}

            bool openChannel(const std::string& cid,
            ConsumerManager::ptr consumerManagerP,
            muduo::net::TcpConnectionPtr connectionP,
            ProtobufCodecPtr codecP,
            Virtualhost::ptr vHost,
            ThreadPool::ptr pool){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _channels.find(cid);
                if(it != _channels.end()){
                    return false;
                }
                Channel::ptr cp = std::make_shared<Channel>(cid,consumerManagerP,connectionP,
                codecP,vHost,pool);
                _channels.insert(std::make_pair(cid,cp));
                return true;
            }
            void closeChannel(const std::string& cid){
                std::unique_lock<std::mutex> lock(_mutex);
                _channels.erase(cid);
            }
            Channel::ptr getChannel(const std::string& cid){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _channels.find(cid);
                if(it == _channels.end()){
                    return Channel::ptr();
                }
                return it->second;
            }
        private:
            std::mutex _mutex;
            std::unordered_map<std::string,Channel::ptr> _channels;
    };
    
}

#endif